package com.doit.day01;

import org.apache.commons.lang3.RandomStringUtils;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

public class ProducerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        /**
         * 下面的参数是必须要配置的
         */
        //配置kafka集群地址
//        props.setProperty("bootstrap.servers","linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        //key的序列化方式，必须是org.apache.kafka.common.serialization.Serializer接口的实现类
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //org.apache.kafka.common.serialization.StringSerializer
//        props.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        props.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer");


        /**
         * 下面是可选配的一些重要参数
         */
        //发送失败重试的次数
//        props.setProperty("retries","4");
        //生产者的应答机制
        props.setProperty("acks","-1");
        //partitioner.class  分区器   发送一条消息，默认是随机发送的
        props.setProperty("partitioner.class","com.doit.day01.MyPartitioner");
        //interceptor.classes  ==》 拦截器  ==》
        props.setProperty("interceptor.classes","com.doit.day01.MyInterceptor");
        //transactional.id  事务id

        //创建kafka生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);


        for (int i = 0; i < 1000; i++) {
            //构造数据
            ProducerRecord<String, String> record = new ProducerRecord<>("test", "abcde"+i);
            //发送数据
            producer.send(record);
            //刷新一下
            producer.flush();
        }

        //关闭资源
        producer.close();



    }
}


